package com.mingserve.common.mqtt.utils;

import com.mingserve.common.mqtt.config.properties.MqttProperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 线程安全的MQTT客户端管理工具类
 * 特性：
 * 1. 支持预初始化（init）和懒加载（getInstance）两种模式
 * 2. 自动重连（AutomaticReconnect）
 * 3. 无锁化线程安全（ConcurrentHashMap + computeIfAbsent）
 * 4. 完善的异常处理和日志
 *
 * @author lvwshuai
 * @date 2025-04-01
 */
@Slf4j
@Component
public class MqttUtils {
    // 保存所有 MQTT 客户端实例
    private final Map<String, MqttClient> clientInstances = new ConcurrentHashMap<>();
    // 专门用于同步的锁对象
    private final Object lock = new Object();

    @Autowired
    private MqttProperties configurations;

    /**
     * 初始化所有客户端（应用启动时调用）
     * 注意：已存在的客户端不会被重复初始化
     */
    public void init() {
        for (MqttProperties.ClientConfig config : configurations.getList()) {
            String clientName = config.getClientName();
            try {
                synchronized (lock) {
                    if (!clientInstances.containsKey(clientName)) {
                        MqttClient client = createAndConnectClient(config, clientName);
                        clientInstances.put(clientName, client);
                        log.info("MQTT client {} pre-init successful", clientName);
                    }
                }
            } catch (MqttException e) {
                log.error("MQTT client pre-init failed: {}", clientName, e);
            }
        }
    }

    /**
     * 获取客户端实例（线程安全懒加载）
     * @param clientName 客户端名称（对应配置中的clientName）
     * @return 已连接的MQTT客户端
     * @throws MqttException 如果客户端配置不存在或连接失败
     */
    public MqttClient getInstance(String clientName) throws MqttException {
        MqttClient client = clientInstances.get(clientName);
        if (client == null) { // 只需检查是否存在，不检查连接状态！ AutomaticReconnect 设为 true
            synchronized (lock) { // 使用专用锁对象，避免锁 this
                client = clientInstances.get(clientName);
                if (client == null) {
                    MqttProperties.ClientConfig config = configurations.getList().stream().filter(c -> c.getClientName().equals(clientName)).findFirst().orElseThrow(() -> new IllegalArgumentException("未找到MQTT客户端配置: " + clientName));
                    client = createAndConnectClient(config, clientName);
                    clientInstances.put(clientName, client);
                    log.info("MQTT client {} lazy loading init successful", clientName);
                }
            }
        }
        return client;
    }

    /**
     * 创建并连接MQTT客户端（内部方法）
     */
    private MqttClient createAndConnectClient(MqttProperties.ClientConfig config, String clientName) throws MqttException {
        MqttClient client = new MqttClient(config.getUrl(), config.getClientId(), new MemoryPersistence());

        // 设置回调（包含连接状态监听）
        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("MQTT client {} {}connection successful", clientName, reconnect ? "re-" : "");
            }

            @Override
            public void connectionLost(Throwable cause) {
                log.warn("MQTT client {} connection lost: {}", clientName,  cause != null ? cause.getMessage() : "网络不可用");
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                log.debug("MQTT client {} received msg {}", clientName, topic);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.debug("MQTT client {} msg deliver completed", clientName);
            }
        });

        client.connect(getMqttConnectOptions(config));
        return client;
    }

    /**
     *  构建MQTT连接选项（自动重连+持久会话）
     *
     * @param config
     * @return
     */
    private MqttConnectOptions getMqttConnectOptions(MqttProperties.ClientConfig config) {
        MqttConnectOptions options = new MqttConnectOptions();
        // 自动重连，移除 @Scheduled 的 checkMqttConnection 方法
        options.setAutomaticReconnect(true);
        options.setMaxReconnectDelay(30000); // 最大重连间隔 30秒
        // 手动检测，自动重连与定时检测有冲突，改为false
//        options.setAutomaticReconnect(false);
        options.setCleanSession(false);
        options.setConnectionTimeout(config.getConnectionTimeOut());
        options.setKeepAliveInterval(config.getKeepAlive());
        options.setUserName(config.getUsername());
        options.setPassword(config.getPassword().toCharArray());
        options.setServerURIs(new String[]{config.getUrl()});
        options.setMaxInflight(100); // 设置最大并发发布数
        return options;
    }

    /**
     * 关闭所有 MQTT 客户端连接
     */
    public void shutdown() {
        clientInstances.forEach((clientName, client) -> {
            try {
                if (client.isConnected()) {
                    client.disconnect();
                }
                client.close();
                log.info("MQTT client {} has been disconnected", clientName);
            } catch (MqttException e) {
                log.error("MQTT client {} disconnection failed: ", clientName, e);
            }
        });
        clientInstances.clear(); // 清除所有客户端
    }

    /**
     * 发布消息（自动处理客户端初始化）- 同步
     *
     * @param clientName MQTT 客户端名称
     * @param topic      主题名称
     * @param message    消息内容，qos 默认为1
     */
    public void publish(String clientName, String topic, String message) {
        publish(clientName, topic, message, 1, false);
    }

    /**
     * 发布消息（自动处理客户端初始化）- 同步
     *
     * @param clientName MQTT 客户端名称
     * @param topic      主题名称
     * @param message    消息内容
     * @param qos        QoS=1（至少一次）或QoS=2（精确一次）
     * @param retain     默认false 保留消息 存储最后一条被标记为retained=true的消息
     */
    public void publish(String clientName, String topic, String message, int qos, boolean retain) {
        try {
            MqttClient mqttClient = getInstance(clientName);
            MqttMessage mqttMessage = new MqttMessage(message.getBytes());
            mqttMessage.setQos(qos);
            mqttMessage.setRetained(retain);
            mqttClient.publish(topic, mqttMessage);
            log.info("MQTT {} [{}] msg published successful", clientName, topic);
        } catch (Exception e) {
            log.error("MQTT {} [{}] msg publishing failed", clientName, topic, e);
        }
    }

    /**
     * 发布消息（自动处理客户端初始化）- 异步
     * 需要高吞吐且能容忍少量消息延迟
     *
     * @param clientName MQTT 客户端名称
     * @param topic      主题名称
     * @param message    消息内容
     */
    @Async
    public void publishAsync(String clientName, String topic, String message) {
        publish(clientName, topic, message);
    }

    /**
     * 订阅主题（自动处理客户端初始化）
     *
     * @param clientName MQTT 客户端名称
     * @param topic      主题名称
     * @param listener   消息监听器
     */
    public void subscribe(String clientName, String topic, IMqttMessageListener listener) {
        try {
            MqttClient mqttClient = getInstance(clientName);
            mqttClient.subscribe(topic, listener);
            log.info("MQTT {} [{}] subscribed successful", clientName, topic);
        } catch (Exception ex) {
            log.error("MQTT {} [{}] subscribing failed", clientName, topic, ex);
        }
    }

//    // 获取CPU核心数
//    int corePoolSize = Runtime.getRuntime().availableProcessors();
//    // 创建一个线程池，用于控制异步发布消息，低优先级线程池（允许堆积）
//    private final ExecutorService mqttExecutor = new ThreadPoolExecutor(
//            corePoolSize,  // 核心线程
//            corePoolSize * 2, // 最大线程
//            60L, TimeUnit.SECONDS,
//            new LinkedBlockingQueue<>(100),  // 扩大队列
//            new ThreadPoolExecutor.CallerRunsPolicy()  // 队列满后由调用线程执行
//    );
//    public void publishAsync(String clientName, String source, String message) {
//        MqttClient mqttClient = clientInstances.get(clientName);
//        if (mqttClient == null) {
//            log.error("MQTT {} [{}] 客户端找不到，发送失败", clientName, source);
//            return;
//        }
//        // 使用线程池异步发送消息
//        mqttExecutor.submit(() -> {
//            try {
//                mqttClient.publish(source, new MqttMessage(message.getBytes()));
//                log.info("MQTT {} [{}] 发送成功", clientName, source);
//            } catch (MqttException e) {
//                log.error("MQTT {} [{}] 发送失败", clientName, source, e);
//            }
//        });
//    }

    /**
     * 定时任务，检测 MQTT 连接状态
     * 每隔一段时间检查 MQTT 客户端是否在线，如果不在线，则重连
     */
//    @Scheduled(fixedRate = 60000, initialDelay = 60000) // 每隔60秒执行一次
    public void checkMqttConnection() {
        log.info("Check MQTT connection");
        clientInstances.forEach((clientName, mqttClient) -> {
            if (mqttClient != null && !mqttClient.isConnected()) {
                log.warn("MQTT client '{}' is offline，trying to reconnect...", clientName);
                try {
//                    mqttClient.reconnect();
                    MqttProperties.ClientConfig config = getConfigByName(clientName); // 需实现根据名称获取配置
                    reconnect(mqttClient, clientName, config);
                } catch (Exception e) {
                    log.error("MQTT client '{}' reconnect failed", clientName, e);
                }
            }
        });
    }

    /**
     * 根据客户端名称获取配置
     *
     * @param clientName
     * @return
     */
    private MqttProperties.ClientConfig getConfigByName(String clientName) {
        return configurations.getList().stream().filter(c -> c.getClientName().equals(clientName)).findFirst().orElseThrow(() -> new IllegalArgumentException("未找到客户端配置: " + clientName));
    }

    /**
     * 连接或重连 MQTT 客户端（公共方法）
     */
    private void reconnect(MqttClient mqttClient, String clientName, MqttProperties.ClientConfig config) {
        try {
            // 1. 强制断开旧连接（如果是重连）
            if (mqttClient.isConnected()) {
                mqttClient.disconnectForcibly();
            }
            // 2. 使用相同的配置重新连接
            MqttConnectOptions options = getMqttConnectOptions(config);
            mqttClient.connect(options);
            log.info("MQTT client '{}' reconnection successful", clientName);
        } catch (MqttException e) {
            log.error("MQTT client '{}' reconnection failed", clientName, e);
        }
    }
}